package com.greenline.push.service.core.task;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.ResponseHandler;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.greenline.push.core.enums.Channels;
import com.greenline.push.core.enums.MsgSendType;
import com.greenline.push.core.enums.TaskStatus;
import com.greenline.push.core.enums.Tasks;
import com.greenline.push.core.pushcenter.ClassDefineManager;
import com.greenline.push.core.pushcenter.MsgForSendManager;
import com.greenline.push.core.pushcenter.MsgRetryFullManager;
import com.greenline.push.core.pushcenter.TaskSyncManager;
import com.greenline.push.dal.pushcenter.dataobject.ClassDefineDO;
import com.greenline.push.dal.pushcenter.dataobject.MsgForSendDO;
import com.greenline.push.dal.pushcenter.dataobject.MsgRetryDO;
import com.greenline.push.dal.pushcenter.dataobject.TaskSyncDO;
import com.greenline.push.service.core.exception.TaskException;
import com.greenline.push.service.core.item.AppPacket;
import com.greenline.push.service.core.item.Packet;
import com.greenline.push.service.core.send.base.AppPacketBuilder;
import com.greenline.push.service.core.tmpl.AppChannelCache;
import com.greenline.push.service.core.tmpl.AppTemplateCache;
import com.greenline.push.service.core.tmpl.AppTemplateInfo;
import com.greenline.push.service.core.user.UserManager;
import com.greenline.push.service.core.utils.Constants;
import com.greenline.push.service.core.utils.PacketConvertor;

public class FetchFilterTypeToRetrySchedule extends TaskBase implements Runnable {

	@Autowired
	private TaskSyncManager taskSyncManager;
	@Autowired
	PacketConvertor convertor;
	@Autowired
	private MsgRetryFullManager msgRetryFullManager;
	@Autowired
	private MsgForSendManager msgForSendManager;
	@Resource
	private AppPacketBuilder appPacketBuilder;
	@Resource
	private UserManager userManager;
	@Autowired
	private AppTemplateCache appTemplateCache;
	@Resource
	private AppChannelCache appChannelCache;
	@Resource
	private ClassDefineManager classManager;

	private TaskSyncDO task = null;
	private Tasks taskType = Tasks.FETCH_FILTER_TO_RETRY;
	private Long lastRecordId = null;
	private final Integer MAX_MSG_COUNT = 100;
	private static final String taskName = "{筛选分组解析到msg_retry_full}";
	private String taskId = null;
	private Integer READ_MAX_PAGE = 20;
	private Logger log = Constants.LOG_SCHEDULE;

	public void startTask() throws TaskException {
		log.info(String.format("%s--任务类型:<%s>开始启动----------", taskName, task));
		// task init from here
		ScheduledExecutorService schedule = Executors.newScheduledThreadPool(1);
		// get period
		long period = taskSyncManager.getTaskSyncByType(taskType.getType()).getRunPeriod();
		// start task
		schedule.scheduleAtFixedRate(this, 0, period, TimeUnit.SECONDS);
		log.info(String.format("%s 任务类型:<%s>启动完成,运行周期:<%s>=========", taskName, taskType, period));
	}

	@Override
	public void run() {
		excuteTask(new TaskCallBack() {
			@Override
			public TaskInfo initTaskInfo() {
				task = taskSyncManager.getTaskSyncByType(taskType.getType());
				// get uuid
				taskId = UUID.randomUUID().toString();
				log.debug(String.format("------定时任务开始执行----taskType:%s,tasks:%s,taskId:%s", taskType.getType(),
						taskType, taskId));
				TaskInfo taskInfo = new TaskInfo(taskName);

				// 无任务或者任务状态为running（status=1为running），task不能启动
				if (task == null) {
					// here ,this return will broken the task forever.
					log.debug(taskType + "-没有找到任务,停止执行。" + taskId);
					return null;
				} else if (task.getStatus() == TaskStatus.RUNNING.getStatus()) {
					log.debug(taskType + "-任务碰撞，停止执行。" + taskId);
					taskInfo.setBizResult(TaskStatus.RUNNING.getName());
					return taskInfo;
				} else {
					task.setStatus(TaskStatus.RUNNING.getStatus());
					taskSyncManager.update(task);

					taskInfo.setBizResult(TaskStatus.END.getName());
					return taskInfo;
				}
			}

			@Override
			public TaskInfo excute(TaskInfo taskInfo) {

				if (task == null) {
					taskInfo.setSeccusss(false);
					taskInfo.setBizResult("失败");
					return taskInfo;
				}
				Long startId = task.getLastRecordId();
				lastRecordId = startId;
				// get filter type push data
				List<MsgForSendDO> msgList = msgForSendManager.fetchByTypeAndRecord(MsgSendType.CONDITIONS.getType(),
						lastRecordId, MAX_MSG_COUNT);

				// deseriziable to packet
				for (MsgForSendDO msgForSendDO : msgList) {
					Integer readerIndex = null;
					/*
					 * if readindex is null or zero ,fetch from 0.other should
					 * fetch from readerindex+1;
					 */
					if (msgForSendDO.getReadIndex() == null || msgForSendDO.getReadIndex() == 0) {
						msgForSendDO.setReadIndex(0);
						readerIndex = 0;
					} else {
						readerIndex = msgForSendDO.getReadIndex() + 1;
					}

					/**
					 * retrieve data from search interface
					 */
					Long classId = msgForSendDO.getClassId();
					ClassDefineDO classDO = classManager.getClassDefineById(classId);
					String conditions = classDO.getConditions();
					List<Long> uidList = getUidListFromSearch(conditions);
					Packet packet = convertor.convertForSendDOToPacket(msgForSendDO);
					List<AppTemplateInfo> appTmpls = appTemplateCache.get(packet.getTemplateId(), packet.getDomainId());
					if (null == appTmpls) {
						log.error(String.format("%s:<<该域下未找到对应的模板:domain:%s.数据会被直接删除", taskName, packet.getDomainId()));
						// delete msg_for_send
						msgForSendManager.delete(msgForSendDO.getId());
						continue;
					}
					int priority = -1;
					for (AppTemplateInfo appTempl : appTmpls) {

						// 优先级比上次处理的大（第一次或者前次推送失败）或者相等就调用推送
						if (appTempl.getPriority() >= priority) {
							// 获取用户 这个循环最多走20次,可以取2000个用户
							for (int i = 0; i < MAX_MSG_COUNT; i++) {
								// List<Long> userIds =
								// userManager.getUserListByAppId(appTempl.getAppId(),
								// readerIndex);
								List<MsgRetryDO> msgRetryList = new ArrayList<MsgRetryDO>();
								for (Long userId : uidList) {
									Channels channel = appChannelCache.getChannel(appTempl.getAppChannelId());
									// build packet
									packet.setType(MsgSendType.CONDITIONS.getType());
									packet.setRecvUserId(userId);
									AppPacket appPacket = appPacketBuilder
											.buildChannelPacket(packet, appTempl, channel);
									// packet 创建成功入库.失败选择低优先级
									if (appPacket == null) {
										continue;
									} else {
										// insert to msg_retry_full
										MsgRetryDO msgRetryDO = wrapPacket(appTempl, channel, appPacket);
										msgRetryList.add(msgRetryDO);
									}
								}

								/*
								 * if userid is null or build packet is
								 * null,search next app template
								 */
								if (msgRetryList.size() > 0) {
									priority = appTempl.getPriority();
								}

								// batch insert to db,each has max 100 records
								if (msgRetryList.size() > 0)
									msgRetryFullManager.batchInsert(msgRetryList);
								if ((readerIndex < READ_MAX_PAGE + msgForSendDO.getReadIndex())
										&& (uidList.size() == UserManager.DEFAULT_PAGE_SIZE)) {
									readerIndex++;
								} else if (uidList.size() < UserManager.DEFAULT_PAGE_SIZE) {
									// 数据已经读完
									msgForSendManager.delete(msgForSendDO.getId());
									break;
								} else {
									log.debug(String.format("已经读取%s页数据.执行结束", readerIndex));
									// update msg_for_send
									msgForSendDO.setReadIndex(readerIndex);
									msgForSendManager.update(msgForSendDO);
									break;
								}

							}
						}
					}
					lastRecordId = msgForSendDO.getId();
				}

				// if fetch data &lt max_count means lastRecordId runs to the
				// last,then should return from zero
				if (msgList.size() < MAX_MSG_COUNT) {
					// if insert all the packets to queue,so we can set
					// lastRecordId to zero;
					if (msgList.size() == 0 || lastRecordId == msgList.get(msgList.size() - 1).getId()) {
						// set
						lastRecordId = 0l;
					}
				}
				// update status
				task.setLastRecordId(lastRecordId);
				task.setStatus(TaskStatus.END.getStatus());
				taskSyncManager.update(task);
				log.debug(String.format("%s--%s:--任务执行结束***任务id:%s", taskName, taskType, taskId));
				// update taskinfo
				taskInfo.setSeccusss(true);
				taskInfo.setCount((long) msgList.size());
				taskInfo.setBizResult("成功");
				return taskInfo;
			}

			//
			private MsgRetryDO wrapPacket(AppTemplateInfo appTempl, Channels channel, AppPacket appPacket) {
				MsgRetryDO msgRetryDO = new MsgRetryDO();
				msgRetryDO.setAppChannel(channel.getType());
				msgRetryDO.setAppId(appTempl.getAppId());
				msgRetryDO.setFailCount(0);
				msgRetryDO.setMemo("筛选推送入库成功");// 定时任务会抓取status=失败的数据
				msgRetryDO.setFirstOfferStatus(1);
				msgRetryDO.setPacketJson(JSON.toJSONString(appPacket));
				msgRetryDO.setTemplateId(appTempl.getTemplateId());
				return msgRetryDO;
			}

			/**
			 * get uidList from search side.
			 * 
			 * @param conditions
			 * @return
			 */
			private List<Long> getUidListFromSearch(String conditions) {
				List<Long> uidList = new ArrayList<Long>();
				return uidList;
			}
		});
	}

	public final static void main(String[] args) throws Exception {
		CloseableHttpClient httpclient = HttpClients.createDefault();
		String url = "http://localhost/";
		try {
			HttpGet httpget = new HttpGet(url);

			System.out.println("Executing request " + httpget.getRequestLine());

			// Create a custom response handler
			ResponseHandler<String> responseHandler = new ResponseHandler<String>() {

				public String handleResponse(final HttpResponse response) throws ClientProtocolException, IOException {
					int status = response.getStatusLine().getStatusCode();
					if (status >= 200 && status < 300) {
						HttpEntity entity = response.getEntity();
						return entity != null ? EntityUtils.toString(entity) : null;
					} else {
						throw new ClientProtocolException("Unexpected response status: " + status);
					}
				}

			};
			String responseBody = httpclient.execute(httpget, responseHandler);
			System.out.println("----------------------------------------");
			System.out.println(responseBody);
		} finally {
			httpclient.close();
		}
	}

}
